package cn.doitedu.kafka.topic;

import kafka.server.KafkaConfig;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.KafkaFuture;
import org.apache.zookeeper.server.ServerConfig;

import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/***
 * @author hunter.d
 * @qq 657270652
 * @wx haitao-duan
 * @date 2020/11/14
 **/
public class KafkaAdminDemo {

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        // 参数配置
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,"doitedu01:9092,doitedu02:9092,doitedu03:9092");
        props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG,3000);

        // 创建 admin client 对象
        AdminClient adminClient = KafkaAdminClient.create(props);

        // 由服务端controller自行分配分区及副本所在broker
        NewTopic tpc_3 = new NewTopic("tpc_3", 2, (short) 1);

        // 手动指定分区及副本的broker分配
        HashMap<Integer, List<Integer>> replicaAssignments = new HashMap<>();
        // 分区0，分配到broker0，broker1
        replicaAssignments.put(0,Arrays.asList(0,1));
        // 分区1，分配到broker0,broker2
        replicaAssignments.put(0,Arrays.asList(0,1));



        NewTopic tpc_4 = new NewTopic("tpc_4", replicaAssignments);
        CreateTopicsResult result = adminClient.createTopics(Arrays.asList(tpc_3,tpc_4));

        // 从future中等待服务端返回
        try {
            result.all().get();
        } catch (Exception e) {
            e.printStackTrace();
        }

        // 删除主题
        DeleteTopicsResult deleteTopicsResult = adminClient.deleteTopics(Arrays.asList("tpc_1", "tpc_1"));
        Map<String, KafkaFuture<Void>> values = deleteTopicsResult.values();
        System.out.println(values);


        // 列出主题
        ListTopicsResult listTopicsResult = adminClient.listTopics();
        Set<String> topics = listTopicsResult.names().get();
        System.out.println(topics);

        // describ 主题
        DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Arrays.asList("tpc_4", "tpc_3"));
        Map<String, TopicDescription> res = describeTopicsResult.all().get();
        Set<String> ksets = res.keySet();
        for (String k : ksets) {
            System.out.println(res.get(k));
        }


        adminClient.close();
    }

}
